#include "threadpool.h"
#include <sys/epoll.h>
#include <sys/fcntl.h>
#include <sys/sysinfo.h>
#include <unistd.h>

#include <errno.h>
#include <string.h>
#include "log.h"

Task::Task(std::function<void (void*)> callBackFun, void* arg) 
{
	mFTaskRunCB = callBackFun;
    m_arg = arg;
}

Task::~Task() {}

void Task::run() 
{
	mFTaskRunCB(m_arg);
}

ThreadPool::ThreadPool() 
{
    m_task_size_limit = -1;	// means unlimit
    m_pool_size = 0;
    m_pool_state = -1;
}

ThreadPool::~ThreadPool()
{
    // Release resources
    if (m_pool_state != STOPPED) 
	{
        destroy_threadpool();
    }
}

void ThreadPool::StartThread(void* arg)
{
    ThreadPool* tp = (ThreadPool *)arg;
    if (tp->mFStartedCB) 
	{
        tp->mFStartedCB();
    } 
	else 
	{
        DEBUG_MSG("thread start cb is null");
    }

    tp->execute_thread();
}

int ThreadPool::start()
{
    if (m_pool_size == 0) {
        ERROR_MSG("pool size must be set!");
        return -1;
    }
    if (m_pool_state == STARTED) 
	{
        WARNING_MSG("ThreadPool has started!");
        return 0;
    }
    m_pool_state = STARTED;
	for (int i = 0 ; i < m_pool_size; ++i)
	{
		allThreads.push_back(std::thread(&ThreadPool::StartThread, this, this));
	}

    return 0;
}

void ThreadPool::set_thread_start_cb(std::function<void(void)> startedCB) 
{
	mFStartedCB = startedCB;
}

void ThreadPool::set_task_size_limit(int size) 
{
    m_task_size_limit = size;
}

void ThreadPool::set_pool_size(int pool_size) 
{
    m_pool_size = pool_size;
}

int ThreadPool::destroy_threadpool()
{
    // Note: this is not for synchronization, its for thread communication!
    // destroy_threadpool() will only be called from the main thread, yet
    // the modified m_pool_state may not show up to other threads until its 
    // modified in a lock!

	m_task_mutex.lock();

	DEBUG_MSG("locking: " << std::this_thread::get_id());
	m_pool_state = STOPPED;
	m_task_mutex.unlock();

    INFO_MSG("Broadcasting STOP signal to all threads...");
    m_task_cond_var.notify_all(); // notify all threads we are shttung down

	for (auto &m_thread : allThreads)
	{
		if (m_thread.joinable())
		{
			m_thread.join();	
			DEBUG_MSG("thread ID: " << m_thread.get_id() << " joined");
			m_task_cond_var.notify_all(); // try waking up a bunch of threads that are still waiting
		}
		else
		{
			ERROR_MSG("thread ID: " << m_thread.get_id() << " can not join");
		}
	}

    INFO_MSG(m_pool_size << " threads exited from the thread pool, task size: " <<  m_tasks.size());
    return 0;
}

void* ThreadPool::execute_thread()
{
	Task *task;
	INFO_MSG("created thread : " << std::this_thread::get_id());
	while(m_pool_state != STOPPED) 
	{
		m_task_mutex.lock();
		m_task_cond_var.wait(m_task_mutex);

		// If the thread was woken up to notify process shutdown, return from here
		if (m_pool_state == STOPPED) 
		{
			INFO_MSG("Unlocking and exiting: " << std::this_thread::get_id());
			return nullptr;
		}

		while(!m_tasks.empty())
		{
			task = m_tasks.front();
			task->run(); // execute the task
			delete task;
			m_tasks.pop();
		}

		m_task_mutex.unlock();
	}

	INFO_MSG("exit thread : " << std::this_thread::get_id());
	return nullptr;
}

int ThreadPool::add_task(Task *task)
{
	m_task_mutex.lock();

    if (m_task_size_limit > 0 && (int) m_tasks.size() > m_task_size_limit) 
	{
        WARNING_MSG("task size reach limit: " << m_task_size_limit);
        m_task_mutex.unlock();
        return -1;
    }

    m_tasks.push(task);

    m_task_cond_var.notify_one(); // wake up one thread that is waiting for a task to be available

	m_task_mutex.unlock();

    return 0;
}
